{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "c4419168-c0e6-4a65-b56e-8454c42060ac",
   "metadata": {
    "tags": []
   },
   "source": [
    "### 0. Spark Setup"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "32bd7cdd-8504-4a54-a461-244bf7878d2a",
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.3.1,org.apache.spark:spark-avro_2.12:3.3.1 pyspark-shell'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "3aab2a7e-a685-4925-9c9a-b5adf201af77",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "your 131072x1 screen size is bogus. expect trouble\n",
      "24/03/11 00:28:48 WARN Utils: Your hostname, Cinders resolves to a loopback address: 127.0.1.1; using 172.17.156.62 instead (on interface eth0)\n",
      "24/03/11 00:28:48 WARN Utils: Set SPARK_LOCAL_IP if you need to bind to another address\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      ":: loading settings :: url = jar:file:/home/ellabelle/spark/spark-3.5.1-bin-hadoop3/jars/ivy-2.5.1.jar!/org/apache/ivy/core/settings/ivysettings.xml\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Ivy Default Cache set to: /home/ellabelle/.ivy2/cache\n",
      "The jars for the packages stored in: /home/ellabelle/.ivy2/jars\n",
      "org.apache.spark#spark-sql-kafka-0-10_2.12 added as a dependency\n",
      "org.apache.spark#spark-avro_2.12 added as a dependency\n",
      ":: resolving dependencies :: org.apache.spark#spark-submit-parent-0c8615d6-fa19-46ec-942b-46e9fe0012aa;1.0\n",
      "\tconfs: [default]\n",
      "\tfound org.apache.spark#spark-sql-kafka-0-10_2.12;3.3.1 in central\n",
      "\tfound org.apache.spark#spark-token-provider-kafka-0-10_2.12;3.3.1 in central\n",
      "\tfound org.apache.kafka#kafka-clients;2.8.1 in central\n",
      "\tfound org.lz4#lz4-java;1.8.0 in central\n",
      "\tfound org.xerial.snappy#snappy-java;1.1.8.4 in central\n",
      "\tfound org.slf4j#slf4j-api;1.7.32 in central\n",
      "\tfound org.apache.hadoop#hadoop-client-runtime;3.3.2 in central\n",
      "\tfound org.spark-project.spark#unused;1.0.0 in central\n",
      "\tfound org.apache.hadoop#hadoop-client-api;3.3.2 in central\n",
      "\tfound commons-logging#commons-logging;1.1.3 in central\n",
      "\tfound com.google.code.findbugs#jsr305;3.0.0 in central\n",
      "\tfound org.apache.commons#commons-pool2;2.11.1 in central\n",
      "\tfound org.apache.spark#spark-avro_2.12;3.3.1 in central\n",
      "\tfound org.tukaani#xz;1.8 in central\n",
      ":: resolution report :: resolve 328ms :: artifacts dl 13ms\n",
      "\t:: modules in use:\n",
      "\tcom.google.code.findbugs#jsr305;3.0.0 from central in [default]\n",
      "\tcommons-logging#commons-logging;1.1.3 from central in [default]\n",
      "\torg.apache.commons#commons-pool2;2.11.1 from central in [default]\n",
      "\torg.apache.hadoop#hadoop-client-api;3.3.2 from central in [default]\n",
      "\torg.apache.hadoop#hadoop-client-runtime;3.3.2 from central in [default]\n",
      "\torg.apache.kafka#kafka-clients;2.8.1 from central in [default]\n",
      "\torg.apache.spark#spark-avro_2.12;3.3.1 from central in [default]\n",
      "\torg.apache.spark#spark-sql-kafka-0-10_2.12;3.3.1 from central in [default]\n",
      "\torg.apache.spark#spark-token-provider-kafka-0-10_2.12;3.3.1 from central in [default]\n",
      "\torg.lz4#lz4-java;1.8.0 from central in [default]\n",
      "\torg.slf4j#slf4j-api;1.7.32 from central in [default]\n",
      "\torg.spark-project.spark#unused;1.0.0 from central in [default]\n",
      "\torg.tukaani#xz;1.8 from central in [default]\n",
      "\torg.xerial.snappy#snappy-java;1.1.8.4 from central in [default]\n",
      "\t---------------------------------------------------------------------\n",
      "\t|                  |            modules            ||   artifacts   |\n",
      "\t|       conf       | number| search|dwnlded|evicted|| number|dwnlded|\n",
      "\t---------------------------------------------------------------------\n",
      "\t|      default     |   14  |   0   |   0   |   0   ||   14  |   0   |\n",
      "\t---------------------------------------------------------------------\n",
      ":: retrieving :: org.apache.spark#spark-submit-parent-0c8615d6-fa19-46ec-942b-46e9fe0012aa\n",
      "\tconfs: [default]\n",
      "\t0 artifacts copied, 14 already retrieved (0kB/8ms)\n",
      "24/03/11 00:28:50 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n",
      "Setting default log level to \"WARN\".\n",
      "To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\n",
      "24/03/11 00:28:51 WARN Utils: Service 'SparkUI' could not bind on port 4040. Attempting port 4041.\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "import pyspark.sql.types as T\n",
    "import pyspark.sql.functions as F\n",
    "\n",
    "spark = SparkSession \\\n",
    "    .builder \\\n",
    "    .appName(\"Spark-Notebook\") \\\n",
    "    .getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6f4b62fa-b3ce-4a1b-a1f4-2ed332a0d55a",
   "metadata": {
    "tags": []
   },
   "source": [
    "### 1. Reading from Kafka Stream\n",
    "\n",
    "through `readStream`"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f491fa45-4471-4bc5-92f7-48081f687140",
   "metadata": {},
   "source": [
    "#### 1.1 Raw Kafka Stream"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "82c25cb2-2599-4f9b-8849-967fbb604a44",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# default for startingOffsets is \"latest\"\n",
    "df_kafka_raw = spark \\\n",
    "    .readStream \\\n",
    "    .format(\"kafka\") \\\n",
    "    .option(\"kafka.bootstrap.servers\", \"localhost:9092,broker:29092\") \\\n",
    "    .option(\"subscribe\", \"rides_csv\") \\\n",
    "    .option(\"startingOffsets\", \"earliest\") \\\n",
    "    .option(\"checkpointLocation\", \"checkpoint\") \\\n",
    "    .load()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "d9149ccd-69b2-4f5b-afc0-43567673c634",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- key: binary (nullable = true)\n",
      " |-- value: binary (nullable = true)\n",
      " |-- topic: string (nullable = true)\n",
      " |-- partition: integer (nullable = true)\n",
      " |-- offset: long (nullable = true)\n",
      " |-- timestamp: timestamp (nullable = true)\n",
      " |-- timestampType: integer (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_kafka_raw.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "62e5e753-89c7-460f-a8be-16868ce5c680",
   "metadata": {
    "tags": []
   },
   "source": [
    "#### 1.2 Encoded Kafka Stream"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "0b745eed-7d74-421e-8e4b-c8343fda4de3",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "df_kafka_encoded = df_kafka_raw.selectExpr(\"CAST(key AS STRING)\",\"CAST(value AS STRING)\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "6839addc-c7c0-4117-8c9c-d2cd59cbf136",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- key: string (nullable = true)\n",
      " |-- value: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_kafka_encoded.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6749c4de-6f80-4b91-b2b8-b2968c761d75",
   "metadata": {},
   "source": [
    "#### 1.3 Structure Streaming DataFrame"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "ca20ae37-49f0-421f-9859-73fac8d4ca45",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "def parse_ride_from_kafka_message(df_raw, schema):\n",
    "    \"\"\" take a Spark Streaming df and parse value col based on <schema>, return streaming df cols in schema \"\"\"\n",
    "    assert df_raw.isStreaming is True, \"DataFrame doesn't receive streaming data\"\n",
    "\n",
    "    df = df_raw.selectExpr(\"CAST(key AS STRING)\", \"CAST(value AS STRING)\")\n",
    "\n",
    "    # split attributes to nested array in one Column\n",
    "    col = F.split(df['value'], ', ')\n",
    "\n",
    "    # expand col to multiple top-level columns\n",
    "    for idx, field in enumerate(schema):\n",
    "        df = df.withColumn(field.name, col.getItem(idx).cast(field.dataType))\n",
    "    return df.select([field.name for field in schema])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "e1737bd0-146f-4ee2-a70f-a4657af5bbc6",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "ride_schema = T.StructType(\n",
    "    [T.StructField(\"vendor_id\", T.IntegerType()),\n",
    "     T.StructField('tpep_pickup_datetime', T.TimestampType()),\n",
    "     T.StructField('tpep_dropoff_datetime', T.TimestampType()),\n",
    "     T.StructField(\"passenger_count\", T.IntegerType()),\n",
    "     T.StructField(\"trip_distance\", T.FloatType()),\n",
    "     T.StructField(\"payment_type\", T.IntegerType()),\n",
    "     T.StructField(\"total_amount\", T.FloatType()),\n",
    "     ])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "ae2ce896-f54b-4166-b01f-b5532ab292fe",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "df_rides = parse_ride_from_kafka_message(df_raw=df_kafka_raw, schema=ride_schema)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "cd848228-97c5-4325-8457-97f35e533cd8",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- vendor_id: integer (nullable = true)\n",
      " |-- tpep_pickup_datetime: timestamp (nullable = true)\n",
      " |-- tpep_dropoff_datetime: timestamp (nullable = true)\n",
      " |-- passenger_count: integer (nullable = true)\n",
      " |-- trip_distance: float (nullable = true)\n",
      " |-- payment_type: integer (nullable = true)\n",
      " |-- total_amount: float (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_rides.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f1cdb53e-f477-4137-8412-6915d7772125",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "df_rides.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "60277fdc-2797-4b23-9ecf-956b76db5778",
   "metadata": {
    "tags": []
   },
   "source": [
    "### 2 Sink Operation & Streaming Query\n",
    "\n",
    "through `writeStream`\n",
    "\n",
    "---\n",
    "**Output Sinks**\n",
    "- File Sink: stores the output to the directory\n",
    "- Kafka Sink: stores the output to one or more topics in Kafka\n",
    "- Foreach Sink:\n",
    "- (for debugging) Console Sink, Memory Sink\n",
    "\n",
    "Further details can be found in [Output Sinks](https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#output-sinks)\n",
    "\n",
    "---\n",
    "There are three types of **Output Modes**:\n",
    "- Complete: The whole Result Table will be outputted to the sink after every trigger. This is supported for aggregation queries.\n",
    "- Append (default): Only new rows are added to the Result Table\n",
    "- Update: Only updated rows are outputted\n",
    "\n",
    "[Output Modes](https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#output-modes) differs based on the set of transformations applied to the streaming data. \n",
    "\n",
    "--- \n",
    "**Triggers**\n",
    "\n",
    "The [trigger settings](https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#triggers) of a streaming query define the timing of streaming data processing. Spark streaming support micro-batch streamings schema and you can select following options based on requirements.\n",
    "\n",
    "- default-micro-batch-mode\n",
    "- fixed-interval-micro-batch-mode\n",
    "- one-time-micro-batch-mode\n",
    "- available-now-micro-batch-mode\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "02ca9b08-aa61-46cd-b946-4457ce2cdf5d",
   "metadata": {
    "tags": []
   },
   "source": [
    "#### Console and Memory Sink"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "74c72469-4c37-417c-a866-a1c1ef75ae8b",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "def sink_console(df, output_mode: str = 'complete', processing_time: str = '5 seconds'):\n",
    "    write_query = df.writeStream \\\n",
    "        .outputMode(output_mode) \\\n",
    "        .trigger(processingTime=processing_time) \\\n",
    "        .format(\"console\") \\\n",
    "        .option(\"truncate\", False) \\\n",
    "        .start()\n",
    "    return write_query # pyspark.sql.streaming.StreamingQuery"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "d866c7ba-f8e9-475d-830a-50ffb2c5472b",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "24/03/11 00:30:31 WARN ResolveWriteToStream: Temporary checkpoint location created which is deleted normally when the query didn't fail: /tmp/temporary-2b8e8845-1369-4653-8c23-c45a98e194a9. If it's required to delete it under any circumstances, please set spark.sql.streaming.forceDeleteTempCheckpointLocation to true. Important to know deleting temp checkpoint folder is best effort.\n",
      "24/03/11 00:30:31 WARN ResolveWriteToStream: spark.sql.adaptive.enabled is not supported in streaming DataFrames/Datasets and will be disabled.\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "24/03/11 00:30:32 WARN ClientUtils: Couldn't resolve server broker:29092 from bootstrap.servers as DNS resolution failed for broker\n",
      "24/03/11 00:30:32 WARN AdminClientConfig: The configuration 'key.deserializer' was supplied but isn't a known config.\n",
      "24/03/11 00:30:32 WARN AdminClientConfig: The configuration 'value.deserializer' was supplied but isn't a known config.\n",
      "24/03/11 00:30:32 WARN AdminClientConfig: The configuration 'enable.auto.commit' was supplied but isn't a known config.\n",
      "24/03/11 00:30:32 WARN AdminClientConfig: The configuration 'max.poll.records' was supplied but isn't a known config.\n",
      "24/03/11 00:30:32 WARN AdminClientConfig: The configuration 'auto.offset.reset' was supplied but isn't a known config.\n",
      "24/03/11 00:30:33 WARN ClientUtils: Couldn't resolve server broker:29092 from bootstrap.servers as DNS resolution failed for broker\n",
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-------------------------------------------\n",
      "Batch: 0\n",
      "-------------------------------------------\n",
      "+---------+--------------------+---------------------+---------------+-------------+------------+------------+\n",
      "|vendor_id|tpep_pickup_datetime|tpep_dropoff_datetime|passenger_count|trip_distance|payment_type|total_amount|\n",
      "+---------+--------------------+---------------------+---------------+-------------+------------+------------+\n",
      "|1        |2020-07-01 00:25:32 |2020-07-01 00:33:39  |1              |1.5          |2           |9.3         |\n",
      "|1        |2020-07-01 00:03:19 |2020-07-01 00:25:43  |1              |9.5          |1           |27.8        |\n",
      "|2        |2020-07-01 00:15:11 |2020-07-01 00:29:24  |1              |5.85         |2           |22.3        |\n",
      "|2        |2020-07-01 00:30:49 |2020-07-01 00:38:26  |1              |1.9          |1           |14.16       |\n",
      "|2        |2020-07-01 00:31:26 |2020-07-01 00:38:02  |1              |1.25         |2           |7.8         |\n",
      "+---------+--------------------+---------------------+---------------+-------------+------------+------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "write_query = sink_console(df_rides, output_mode='append')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "a9bfa73f-a8cc-4988-a8cf-bf31ee6c449c",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "def sink_memory(df, query_name, query_template):\n",
    "    write_query = df \\\n",
    "        .writeStream \\\n",
    "        .queryName(query_name) \\\n",
    "        .format('memory') \\\n",
    "        .start()\n",
    "    query_str = query_template.format(table_name=query_name)\n",
    "    query_results = spark.sql(query_str)\n",
    "    return write_query, query_results"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "b31d0b76-e917-44e7-a14d-f9ce6901c23a",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "24/03/11 00:30:42 WARN ResolveWriteToStream: Temporary checkpoint location created which is deleted normally when the query didn't fail: /tmp/temporary-c7621425-b7fb-47fe-8b42-791c9c5d3186. If it's required to delete it under any circumstances, please set spark.sql.streaming.forceDeleteTempCheckpointLocation to true. Important to know deleting temp checkpoint folder is best effort.\n",
      "24/03/11 00:30:42 WARN ResolveWriteToStream: spark.sql.adaptive.enabled is not supported in streaming DataFrames/Datasets and will be disabled.\n",
      "24/03/11 00:30:43 WARN ClientUtils: Couldn't resolve server broker:29092 from bootstrap.servers as DNS resolution failed for broker\n",
      "24/03/11 00:30:43 WARN AdminClientConfig: The configuration 'key.deserializer' was supplied but isn't a known config.\n",
      "24/03/11 00:30:43 WARN AdminClientConfig: The configuration 'value.deserializer' was supplied but isn't a known config.\n",
      "24/03/11 00:30:43 WARN AdminClientConfig: The configuration 'enable.auto.commit' was supplied but isn't a known config.\n",
      "24/03/11 00:30:43 WARN AdminClientConfig: The configuration 'max.poll.records' was supplied but isn't a known config.\n",
      "24/03/11 00:30:43 WARN AdminClientConfig: The configuration 'auto.offset.reset' was supplied but isn't a known config.\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "24/03/11 00:30:43 WARN ClientUtils: Couldn't resolve server broker:29092 from bootstrap.servers as DNS resolution failed for broker\n",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "query_name = 'vendor_id_counts'\n",
    "query_template = 'select count(distinct(vendor_id)) from {table_name}'\n",
    "write_query, df_vendor_id_counts = sink_memory(df=df_rides, query_name=query_name, query_template=query_template)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "4ba56111-83bf-4028-ac65-565e0190f310",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<class 'pyspark.sql.streaming.query.StreamingQuery'>\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "{'message': 'Waiting for data to arrive',\n",
       " 'isDataAvailable': False,\n",
       " 'isTriggerActive': False}"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-------------------------------------------\n",
      "Batch: 1\n",
      "-------------------------------------------\n",
      "+---------+--------------------+---------------------+---------------+-------------+------------+------------+\n",
      "|vendor_id|tpep_pickup_datetime|tpep_dropoff_datetime|passenger_count|trip_distance|payment_type|total_amount|\n",
      "+---------+--------------------+---------------------+---------------+-------------+------------+------------+\n",
      "|1        |2020-07-01 00:25:32 |2020-07-01 00:33:39  |1              |1.5          |2           |9.3         |\n",
      "|1        |2020-07-01 00:03:19 |2020-07-01 00:25:43  |1              |9.5          |1           |27.8        |\n",
      "|2        |2020-07-01 00:15:11 |2020-07-01 00:29:24  |1              |5.85         |2           |22.3        |\n",
      "|2        |2020-07-01 00:30:49 |2020-07-01 00:38:26  |1              |1.9          |1           |14.16       |\n",
      "|2        |2020-07-01 00:31:26 |2020-07-01 00:38:02  |1              |1.25         |2           |7.8         |\n",
      "+---------+--------------------+---------------------+---------------+-------------+------------+------------+\n",
      "\n",
      "-------------------------------------------\n",
      "Batch: 2\n",
      "-------------------------------------------\n",
      "+---------+--------------------+---------------------+---------------+-------------+------------+------------+\n",
      "|vendor_id|tpep_pickup_datetime|tpep_dropoff_datetime|passenger_count|trip_distance|payment_type|total_amount|\n",
      "+---------+--------------------+---------------------+---------------+-------------+------------+------------+\n",
      "|1        |2020-07-01 00:25:32 |2020-07-01 00:33:39  |1              |1.5          |2           |9.3         |\n",
      "|1        |2020-07-01 00:03:19 |2020-07-01 00:25:43  |1              |9.5          |1           |27.8        |\n",
      "|2        |2020-07-01 00:15:11 |2020-07-01 00:29:24  |1              |5.85         |2           |22.3        |\n",
      "|2        |2020-07-01 00:30:49 |2020-07-01 00:38:26  |1              |1.9          |1           |14.16       |\n",
      "|2        |2020-07-01 00:31:26 |2020-07-01 00:38:02  |1              |1.25         |2           |7.8         |\n",
      "+---------+--------------------+---------------------+---------------+-------------+------------+------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "print(type(write_query)) # pyspark.sql.streaming.StreamingQuery\n",
    "write_query.status"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "7cc37bda-9cfa-402b-9d42-a6ba5271476b",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------------+\n",
      "|count(DISTINCT vendor_id)|\n",
      "+-------------------------+\n",
      "|                        2|\n",
      "+-------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_vendor_id_counts.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "88862ca9-4d89-487e-987f-08a2b9e83efe",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "write_query.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "443d4041-06db-4a4a-89c1-348848cc7ca8",
   "metadata": {
    "tags": []
   },
   "source": [
    "#### Kafka Sink\n",
    "\n",
    "To write stream results to `kafka-topic`, the stream dataframe has at least a column with name `value`.\n",
    "\n",
    "Therefore before starting `writeStream` in kafka format, dataframe needs to be updated accordingly.\n",
    "\n",
    "More information regarding kafka sink expected data structure [here](https://spark.apache.org/docs/latest/structured-streaming-kafka-integration.html#writing-data-to-kafka)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "8b08a013-d039-41cf-94fd-a1a57571d25f",
   "metadata": {
    "scrolled": true,
    "tags": []
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "24/03/11 00:34:35 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:34:35 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:34:35 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:34:35 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:34:36 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:34:37 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:34:39 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:34:40 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:34:41 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:34:42 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:34:43 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:34:44 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:34:45 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:34:46 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:34:47 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:34:48 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:34:49 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:34:50 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:34:51 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:34:52 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:34:53 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:34:54 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:34:55 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:34:56 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:34:57 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:34:58 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:00 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:01 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:02 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:03 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:04 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:05 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:06 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:07 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:08 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:09 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:10 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:11 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:12 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:13 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:14 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:16 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:17 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:17 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:19 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:20 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:21 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:22 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:23 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:24 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:25 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:26 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:27 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:28 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:29 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:30 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:31 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:32 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:33 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:34 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:35 WARN KafkaOffsetReaderAdmin: Error in attempt 1 getting Kafka offsets: \n",
      "java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TimeoutException: Call(callName=describeTopics, deadlineMs=1710088535000, tries=1, nextAllowedTryMs=1710088535101) timed out at 1710088535001 after 1 attempt(s)\n",
      "\tat org.apache.kafka.common.internals.KafkaFutureImpl.wrapAndThrow(KafkaFutureImpl.java:45)\n",
      "\tat org.apache.kafka.common.internals.KafkaFutureImpl.access$000(KafkaFutureImpl.java:32)\n",
      "\tat org.apache.kafka.common.internals.KafkaFutureImpl$SingleWaiter.await(KafkaFutureImpl.java:89)\n",
      "\tat org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:260)\n",
      "\tat org.apache.spark.sql.kafka010.ConsumerStrategy.retrieveAllPartitions(ConsumerStrategy.scala:66)\n",
      "\tat org.apache.spark.sql.kafka010.ConsumerStrategy.retrieveAllPartitions$(ConsumerStrategy.scala:65)\n",
      "\tat org.apache.spark.sql.kafka010.SubscribeStrategy.retrieveAllPartitions(ConsumerStrategy.scala:102)\n",
      "\tat org.apache.spark.sql.kafka010.SubscribeStrategy.assignedTopicPartitions(ConsumerStrategy.scala:113)\n",
      "\tat org.apache.spark.sql.kafka010.KafkaOffsetReaderAdmin.$anonfun$partitionsAssignedToAdmin$1(KafkaOffsetReaderAdmin.scala:499)\n",
      "\tat org.apache.spark.sql.kafka010.KafkaOffsetReaderAdmin.withRetries(KafkaOffsetReaderAdmin.scala:518)\n",
      "\tat org.apache.spark.sql.kafka010.KafkaOffsetReaderAdmin.partitionsAssignedToAdmin(KafkaOffsetReaderAdmin.scala:498)\n",
      "\tat org.apache.spark.sql.kafka010.KafkaOffsetReaderAdmin.fetchLatestOffsets(KafkaOffsetReaderAdmin.scala:297)\n",
      "\tat org.apache.spark.sql.kafka010.KafkaMicroBatchStream.latestOffset(KafkaMicroBatchStream.scala:132)\n",
      "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$constructNextBatch$4(MicroBatchExecution.scala:491)\n",
      "\tat org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken(ProgressReporter.scala:427)\n",
      "\tat org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken$(ProgressReporter.scala:425)\n",
      "\tat org.apache.spark.sql.execution.streaming.StreamExecution.reportTimeTaken(StreamExecution.scala:67)\n",
      "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$constructNextBatch$2(MicroBatchExecution.scala:490)\n",
      "\tat scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:286)\n",
      "\tat scala.collection.Iterator.foreach(Iterator.scala:943)\n",
      "\tat scala.collection.Iterator.foreach$(Iterator.scala:943)\n",
      "\tat scala.collection.AbstractIterator.foreach(Iterator.scala:1431)\n",
      "\tat scala.collection.IterableLike.foreach(IterableLike.scala:74)\n",
      "\tat scala.collection.IterableLike.foreach$(IterableLike.scala:73)\n",
      "\tat scala.collection.AbstractIterable.foreach(Iterable.scala:56)\n",
      "\tat scala.collection.TraversableLike.map(TraversableLike.scala:286)\n",
      "\tat scala.collection.TraversableLike.map$(TraversableLike.scala:279)\n",
      "\tat scala.collection.AbstractTraversable.map(Traversable.scala:108)\n",
      "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$constructNextBatch$1(MicroBatchExecution.scala:479)\n",
      "\tat scala.runtime.java8.JFunction0$mcZ$sp.apply(JFunction0$mcZ$sp.java:23)\n",
      "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.withProgressLocked(MicroBatchExecution.scala:810)\n",
      "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.constructNextBatch(MicroBatchExecution.scala:475)\n",
      "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$runActivatedStream$2(MicroBatchExecution.scala:268)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken(ProgressReporter.scala:427)\n",
      "\tat org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken$(ProgressReporter.scala:425)\n",
      "\tat org.apache.spark.sql.execution.streaming.StreamExecution.reportTimeTaken(StreamExecution.scala:67)\n",
      "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$runActivatedStream$1(MicroBatchExecution.scala:249)\n",
      "\tat org.apache.spark.sql.execution.streaming.ProcessingTimeExecutor.execute(TriggerExecutor.scala:67)\n",
      "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.runActivatedStream(MicroBatchExecution.scala:239)\n",
      "\tat org.apache.spark.sql.execution.streaming.StreamExecution.$anonfun$runStream$1(StreamExecution.scala:311)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:900)\n",
      "\tat org.apache.spark.sql.execution.streaming.StreamExecution.org$apache$spark$sql$execution$streaming$StreamExecution$$runStream(StreamExecution.scala:289)\n",
      "\tat org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.$anonfun$run$1(StreamExecution.scala:211)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat org.apache.spark.JobArtifactSet$.withActiveJobArtifactState(JobArtifactSet.scala:94)\n",
      "\tat org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.run(StreamExecution.scala:211)\n",
      "Caused by: org.apache.kafka.common.errors.TimeoutException: Call(callName=describeTopics, deadlineMs=1710088535000, tries=1, nextAllowedTryMs=1710088535101) timed out at 1710088535001 after 1 attempt(s)\n",
      "Caused by: org.apache.kafka.common.errors.TimeoutException: Timed out waiting for a node assignment. Call: describeTopics\n",
      "24/03/11 00:35:35 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:36 WARN ClientUtils: Couldn't resolve server broker:29092 from bootstrap.servers as DNS resolution failed for broker\n",
      "24/03/11 00:35:36 WARN AdminClientConfig: The configuration 'key.deserializer' was supplied but isn't a known config.\n",
      "24/03/11 00:35:36 WARN AdminClientConfig: The configuration 'value.deserializer' was supplied but isn't a known config.\n",
      "24/03/11 00:35:36 WARN AdminClientConfig: The configuration 'enable.auto.commit' was supplied but isn't a known config.\n",
      "24/03/11 00:35:36 WARN AdminClientConfig: The configuration 'max.poll.records' was supplied but isn't a known config.\n",
      "24/03/11 00:35:36 WARN AdminClientConfig: The configuration 'auto.offset.reset' was supplied but isn't a known config.\n",
      "24/03/11 00:35:36 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:36 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:36 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:36 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:37 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:37 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:39 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:40 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:41 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:42 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:43 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:44 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:45 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:46 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:47 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:48 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:49 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:50 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:51 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:52 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:53 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:55 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:55 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:57 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:58 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:35:59 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:00 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:01 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:02 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:03 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:04 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:06 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:07 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:08 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:09 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:10 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:11 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:12 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:13 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:14 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:15 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:16 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:17 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:18 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:19 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:20 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:22 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:23 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:24 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:25 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:26 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:27 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:28 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:29 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:30 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:31 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:32 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:33 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:35 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:35 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:36 WARN KafkaOffsetReaderAdmin: Error in attempt 2 getting Kafka offsets: \n",
      "java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TimeoutException: Call(callName=describeTopics, deadlineMs=1710088596058, tries=1, nextAllowedTryMs=1710088596159) timed out at 1710088596059 after 1 attempt(s)\n",
      "\tat org.apache.kafka.common.internals.KafkaFutureImpl.wrapAndThrow(KafkaFutureImpl.java:45)\n",
      "\tat org.apache.kafka.common.internals.KafkaFutureImpl.access$000(KafkaFutureImpl.java:32)\n",
      "\tat org.apache.kafka.common.internals.KafkaFutureImpl$SingleWaiter.await(KafkaFutureImpl.java:89)\n",
      "\tat org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:260)\n",
      "\tat org.apache.spark.sql.kafka010.ConsumerStrategy.retrieveAllPartitions(ConsumerStrategy.scala:66)\n",
      "\tat org.apache.spark.sql.kafka010.ConsumerStrategy.retrieveAllPartitions$(ConsumerStrategy.scala:65)\n",
      "\tat org.apache.spark.sql.kafka010.SubscribeStrategy.retrieveAllPartitions(ConsumerStrategy.scala:102)\n",
      "\tat org.apache.spark.sql.kafka010.SubscribeStrategy.assignedTopicPartitions(ConsumerStrategy.scala:113)\n",
      "\tat org.apache.spark.sql.kafka010.KafkaOffsetReaderAdmin.$anonfun$partitionsAssignedToAdmin$1(KafkaOffsetReaderAdmin.scala:499)\n",
      "\tat org.apache.spark.sql.kafka010.KafkaOffsetReaderAdmin.withRetries(KafkaOffsetReaderAdmin.scala:518)\n",
      "\tat org.apache.spark.sql.kafka010.KafkaOffsetReaderAdmin.partitionsAssignedToAdmin(KafkaOffsetReaderAdmin.scala:498)\n",
      "\tat org.apache.spark.sql.kafka010.KafkaOffsetReaderAdmin.fetchLatestOffsets(KafkaOffsetReaderAdmin.scala:297)\n",
      "\tat org.apache.spark.sql.kafka010.KafkaMicroBatchStream.latestOffset(KafkaMicroBatchStream.scala:132)\n",
      "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$constructNextBatch$4(MicroBatchExecution.scala:491)\n",
      "\tat org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken(ProgressReporter.scala:427)\n",
      "\tat org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken$(ProgressReporter.scala:425)\n",
      "\tat org.apache.spark.sql.execution.streaming.StreamExecution.reportTimeTaken(StreamExecution.scala:67)\n",
      "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$constructNextBatch$2(MicroBatchExecution.scala:490)\n",
      "\tat scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:286)\n",
      "\tat scala.collection.Iterator.foreach(Iterator.scala:943)\n",
      "\tat scala.collection.Iterator.foreach$(Iterator.scala:943)\n",
      "\tat scala.collection.AbstractIterator.foreach(Iterator.scala:1431)\n",
      "\tat scala.collection.IterableLike.foreach(IterableLike.scala:74)\n",
      "\tat scala.collection.IterableLike.foreach$(IterableLike.scala:73)\n",
      "\tat scala.collection.AbstractIterable.foreach(Iterable.scala:56)\n",
      "\tat scala.collection.TraversableLike.map(TraversableLike.scala:286)\n",
      "\tat scala.collection.TraversableLike.map$(TraversableLike.scala:279)\n",
      "\tat scala.collection.AbstractTraversable.map(Traversable.scala:108)\n",
      "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$constructNextBatch$1(MicroBatchExecution.scala:479)\n",
      "\tat scala.runtime.java8.JFunction0$mcZ$sp.apply(JFunction0$mcZ$sp.java:23)\n",
      "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.withProgressLocked(MicroBatchExecution.scala:810)\n",
      "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.constructNextBatch(MicroBatchExecution.scala:475)\n",
      "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$runActivatedStream$2(MicroBatchExecution.scala:268)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken(ProgressReporter.scala:427)\n",
      "\tat org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken$(ProgressReporter.scala:425)\n",
      "\tat org.apache.spark.sql.execution.streaming.StreamExecution.reportTimeTaken(StreamExecution.scala:67)\n",
      "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$runActivatedStream$1(MicroBatchExecution.scala:249)\n",
      "\tat org.apache.spark.sql.execution.streaming.ProcessingTimeExecutor.execute(TriggerExecutor.scala:67)\n",
      "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.runActivatedStream(MicroBatchExecution.scala:239)\n",
      "\tat org.apache.spark.sql.execution.streaming.StreamExecution.$anonfun$runStream$1(StreamExecution.scala:311)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:900)\n",
      "\tat org.apache.spark.sql.execution.streaming.StreamExecution.org$apache$spark$sql$execution$streaming$StreamExecution$$runStream(StreamExecution.scala:289)\n",
      "\tat org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.$anonfun$run$1(StreamExecution.scala:211)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat org.apache.spark.JobArtifactSet$.withActiveJobArtifactState(JobArtifactSet.scala:94)\n",
      "\tat org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.run(StreamExecution.scala:211)\n",
      "Caused by: org.apache.kafka.common.errors.TimeoutException: Call(callName=describeTopics, deadlineMs=1710088596058, tries=1, nextAllowedTryMs=1710088596159) timed out at 1710088596059 after 1 attempt(s)\n",
      "Caused by: org.apache.kafka.common.errors.TimeoutException: Timed out waiting for a node assignment. Call: describeTopics\n",
      "24/03/11 00:36:36 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:37 WARN ClientUtils: Couldn't resolve server broker:29092 from bootstrap.servers as DNS resolution failed for broker\n",
      "24/03/11 00:36:37 WARN AdminClientConfig: The configuration 'key.deserializer' was supplied but isn't a known config.\n",
      "24/03/11 00:36:37 WARN AdminClientConfig: The configuration 'value.deserializer' was supplied but isn't a known config.\n",
      "24/03/11 00:36:37 WARN AdminClientConfig: The configuration 'enable.auto.commit' was supplied but isn't a known config.\n",
      "24/03/11 00:36:37 WARN AdminClientConfig: The configuration 'max.poll.records' was supplied but isn't a known config.\n",
      "24/03/11 00:36:37 WARN AdminClientConfig: The configuration 'auto.offset.reset' was supplied but isn't a known config.\n",
      "24/03/11 00:36:37 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:37 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:37 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:37 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:38 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:38 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:40 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:41 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:42 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:43 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:44 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:45 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:46 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:47 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:47 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:48 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:49 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:50 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:52 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:52 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:54 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:55 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:56 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:57 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:58 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:36:59 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:37:00 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:37:01 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:37:02 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:37:03 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:37:05 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:37:05 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:37:06 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:37:08 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:37:09 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:37:10 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:37:11 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:37:12 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:37:13 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:37:14 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:37:15 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:37:16 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:37:17 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:37:18 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:37:19 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:37:20 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:37:21 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:37:22 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:37:23 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:37:24 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:37:25 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:37:26 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:37:27 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:37:28 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:37:29 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:37:31 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:37:32 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:37:33 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:37:34 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:37:35 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:37:36 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:37:37 WARN KafkaOffsetReaderAdmin: Error in attempt 3 getting Kafka offsets: \n",
      "java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TimeoutException: Call(callName=describeTopics, deadlineMs=1710088657106, tries=1, nextAllowedTryMs=1710088657207) timed out at 1710088657107 after 1 attempt(s)\n",
      "\tat org.apache.kafka.common.internals.KafkaFutureImpl.wrapAndThrow(KafkaFutureImpl.java:45)\n",
      "\tat org.apache.kafka.common.internals.KafkaFutureImpl.access$000(KafkaFutureImpl.java:32)\n",
      "\tat org.apache.kafka.common.internals.KafkaFutureImpl$SingleWaiter.await(KafkaFutureImpl.java:89)\n",
      "\tat org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:260)\n",
      "\tat org.apache.spark.sql.kafka010.ConsumerStrategy.retrieveAllPartitions(ConsumerStrategy.scala:66)\n",
      "\tat org.apache.spark.sql.kafka010.ConsumerStrategy.retrieveAllPartitions$(ConsumerStrategy.scala:65)\n",
      "\tat org.apache.spark.sql.kafka010.SubscribeStrategy.retrieveAllPartitions(ConsumerStrategy.scala:102)\n",
      "\tat org.apache.spark.sql.kafka010.SubscribeStrategy.assignedTopicPartitions(ConsumerStrategy.scala:113)\n",
      "\tat org.apache.spark.sql.kafka010.KafkaOffsetReaderAdmin.$anonfun$partitionsAssignedToAdmin$1(KafkaOffsetReaderAdmin.scala:499)\n",
      "\tat org.apache.spark.sql.kafka010.KafkaOffsetReaderAdmin.withRetries(KafkaOffsetReaderAdmin.scala:518)\n",
      "\tat org.apache.spark.sql.kafka010.KafkaOffsetReaderAdmin.partitionsAssignedToAdmin(KafkaOffsetReaderAdmin.scala:498)\n",
      "\tat org.apache.spark.sql.kafka010.KafkaOffsetReaderAdmin.fetchLatestOffsets(KafkaOffsetReaderAdmin.scala:297)\n",
      "\tat org.apache.spark.sql.kafka010.KafkaMicroBatchStream.latestOffset(KafkaMicroBatchStream.scala:132)\n",
      "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$constructNextBatch$4(MicroBatchExecution.scala:491)\n",
      "\tat org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken(ProgressReporter.scala:427)\n",
      "\tat org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken$(ProgressReporter.scala:425)\n",
      "\tat org.apache.spark.sql.execution.streaming.StreamExecution.reportTimeTaken(StreamExecution.scala:67)\n",
      "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$constructNextBatch$2(MicroBatchExecution.scala:490)\n",
      "\tat scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:286)\n",
      "\tat scala.collection.Iterator.foreach(Iterator.scala:943)\n",
      "\tat scala.collection.Iterator.foreach$(Iterator.scala:943)\n",
      "\tat scala.collection.AbstractIterator.foreach(Iterator.scala:1431)\n",
      "\tat scala.collection.IterableLike.foreach(IterableLike.scala:74)\n",
      "\tat scala.collection.IterableLike.foreach$(IterableLike.scala:73)\n",
      "\tat scala.collection.AbstractIterable.foreach(Iterable.scala:56)\n",
      "\tat scala.collection.TraversableLike.map(TraversableLike.scala:286)\n",
      "\tat scala.collection.TraversableLike.map$(TraversableLike.scala:279)\n",
      "\tat scala.collection.AbstractTraversable.map(Traversable.scala:108)\n",
      "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$constructNextBatch$1(MicroBatchExecution.scala:479)\n",
      "\tat scala.runtime.java8.JFunction0$mcZ$sp.apply(JFunction0$mcZ$sp.java:23)\n",
      "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.withProgressLocked(MicroBatchExecution.scala:810)\n",
      "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.constructNextBatch(MicroBatchExecution.scala:475)\n",
      "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$runActivatedStream$2(MicroBatchExecution.scala:268)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken(ProgressReporter.scala:427)\n",
      "\tat org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken$(ProgressReporter.scala:425)\n",
      "\tat org.apache.spark.sql.execution.streaming.StreamExecution.reportTimeTaken(StreamExecution.scala:67)\n",
      "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$runActivatedStream$1(MicroBatchExecution.scala:249)\n",
      "\tat org.apache.spark.sql.execution.streaming.ProcessingTimeExecutor.execute(TriggerExecutor.scala:67)\n",
      "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.runActivatedStream(MicroBatchExecution.scala:239)\n",
      "\tat org.apache.spark.sql.execution.streaming.StreamExecution.$anonfun$runStream$1(StreamExecution.scala:311)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:900)\n",
      "\tat org.apache.spark.sql.execution.streaming.StreamExecution.org$apache$spark$sql$execution$streaming$StreamExecution$$runStream(StreamExecution.scala:289)\n",
      "\tat org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.$anonfun$run$1(StreamExecution.scala:211)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat org.apache.spark.JobArtifactSet$.withActiveJobArtifactState(JobArtifactSet.scala:94)\n",
      "\tat org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.run(StreamExecution.scala:211)\n",
      "Caused by: org.apache.kafka.common.errors.TimeoutException: Call(callName=describeTopics, deadlineMs=1710088657106, tries=1, nextAllowedTryMs=1710088657207) timed out at 1710088657107 after 1 attempt(s)\n",
      "Caused by: org.apache.kafka.common.errors.TimeoutException: Timed out waiting for a node assignment. Call: describeTopics\n",
      "24/03/11 00:37:37 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n",
      "24/03/11 00:37:38 ERROR MicroBatchExecution: Query [id = 4dfba771-eff7-49e7-a3ff-f1aa03a6e840, runId = 0f86ad02-1d50-487a-97c7-72790d8857d8] terminated with error\n",
      "java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TimeoutException: Call(callName=describeTopics, deadlineMs=1710088657106, tries=1, nextAllowedTryMs=1710088657207) timed out at 1710088657107 after 1 attempt(s)\n",
      "\tat org.apache.kafka.common.internals.KafkaFutureImpl.wrapAndThrow(KafkaFutureImpl.java:45)\n",
      "\tat org.apache.kafka.common.internals.KafkaFutureImpl.access$000(KafkaFutureImpl.java:32)\n",
      "\tat org.apache.kafka.common.internals.KafkaFutureImpl$SingleWaiter.await(KafkaFutureImpl.java:89)\n",
      "\tat org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:260)\n",
      "\tat org.apache.spark.sql.kafka010.ConsumerStrategy.retrieveAllPartitions(ConsumerStrategy.scala:66)\n",
      "\tat org.apache.spark.sql.kafka010.ConsumerStrategy.retrieveAllPartitions$(ConsumerStrategy.scala:65)\n",
      "\tat org.apache.spark.sql.kafka010.SubscribeStrategy.retrieveAllPartitions(ConsumerStrategy.scala:102)\n",
      "\tat org.apache.spark.sql.kafka010.SubscribeStrategy.assignedTopicPartitions(ConsumerStrategy.scala:113)\n",
      "\tat org.apache.spark.sql.kafka010.KafkaOffsetReaderAdmin.$anonfun$partitionsAssignedToAdmin$1(KafkaOffsetReaderAdmin.scala:499)\n",
      "\tat org.apache.spark.sql.kafka010.KafkaOffsetReaderAdmin.withRetries(KafkaOffsetReaderAdmin.scala:518)\n",
      "\tat org.apache.spark.sql.kafka010.KafkaOffsetReaderAdmin.partitionsAssignedToAdmin(KafkaOffsetReaderAdmin.scala:498)\n",
      "\tat org.apache.spark.sql.kafka010.KafkaOffsetReaderAdmin.fetchLatestOffsets(KafkaOffsetReaderAdmin.scala:297)\n",
      "\tat org.apache.spark.sql.kafka010.KafkaMicroBatchStream.latestOffset(KafkaMicroBatchStream.scala:132)\n",
      "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$constructNextBatch$4(MicroBatchExecution.scala:491)\n",
      "\tat org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken(ProgressReporter.scala:427)\n",
      "\tat org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken$(ProgressReporter.scala:425)\n",
      "\tat org.apache.spark.sql.execution.streaming.StreamExecution.reportTimeTaken(StreamExecution.scala:67)\n",
      "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$constructNextBatch$2(MicroBatchExecution.scala:490)\n",
      "\tat scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:286)\n",
      "\tat scala.collection.Iterator.foreach(Iterator.scala:943)\n",
      "\tat scala.collection.Iterator.foreach$(Iterator.scala:943)\n",
      "\tat scala.collection.AbstractIterator.foreach(Iterator.scala:1431)\n",
      "\tat scala.collection.IterableLike.foreach(IterableLike.scala:74)\n",
      "\tat scala.collection.IterableLike.foreach$(IterableLike.scala:73)\n",
      "\tat scala.collection.AbstractIterable.foreach(Iterable.scala:56)\n",
      "\tat scala.collection.TraversableLike.map(TraversableLike.scala:286)\n",
      "\tat scala.collection.TraversableLike.map$(TraversableLike.scala:279)\n",
      "\tat scala.collection.AbstractTraversable.map(Traversable.scala:108)\n",
      "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$constructNextBatch$1(MicroBatchExecution.scala:479)\n",
      "\tat scala.runtime.java8.JFunction0$mcZ$sp.apply(JFunction0$mcZ$sp.java:23)\n",
      "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.withProgressLocked(MicroBatchExecution.scala:810)\n",
      "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.constructNextBatch(MicroBatchExecution.scala:475)\n",
      "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$runActivatedStream$2(MicroBatchExecution.scala:268)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken(ProgressReporter.scala:427)\n",
      "\tat org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken$(ProgressReporter.scala:425)\n",
      "\tat org.apache.spark.sql.execution.streaming.StreamExecution.reportTimeTaken(StreamExecution.scala:67)\n",
      "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$runActivatedStream$1(MicroBatchExecution.scala:249)\n",
      "\tat org.apache.spark.sql.execution.streaming.ProcessingTimeExecutor.execute(TriggerExecutor.scala:67)\n",
      "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.runActivatedStream(MicroBatchExecution.scala:239)\n",
      "\tat org.apache.spark.sql.execution.streaming.StreamExecution.$anonfun$runStream$1(StreamExecution.scala:311)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:900)\n",
      "\tat org.apache.spark.sql.execution.streaming.StreamExecution.org$apache$spark$sql$execution$streaming$StreamExecution$$runStream(StreamExecution.scala:289)\n",
      "\tat org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.$anonfun$run$1(StreamExecution.scala:211)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat org.apache.spark.JobArtifactSet$.withActiveJobArtifactState(JobArtifactSet.scala:94)\n",
      "\tat org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.run(StreamExecution.scala:211)\n",
      "Caused by: org.apache.kafka.common.errors.TimeoutException: Call(callName=describeTopics, deadlineMs=1710088657106, tries=1, nextAllowedTryMs=1710088657207) timed out at 1710088657107 after 1 attempt(s)\n",
      "Caused by: org.apache.kafka.common.errors.TimeoutException: Timed out waiting for a node assignment. Call: describeTopics\n"
     ]
    }
   ],
   "source": [
    "def prepare_dataframe_to_kafka_sink(df, value_columns, key_column=None):\n",
    "    columns = df.columns\n",
    "    df = df.withColumn(\"value\", F.concat_ws(', ',*value_columns))    \n",
    "    if key_column:\n",
    "        df = df.withColumnRenamed(key_column,\"key\")\n",
    "        df = df.withColumn(\"key\",df.key.cast('string'))\n",
    "    return df.select(['key', 'value'])\n",
    "    \n",
    "def sink_kafka(df, topic, output_mode='append'):\n",
    "    write_query = df.writeStream \\\n",
    "        .format(\"kafka\") \\\n",
    "        .option(\"kafka.bootstrap.servers\", \"localhost:9092,broker:29092\") \\\n",
    "        .outputMode(output_mode) \\\n",
    "        .option(\"topic\", topic) \\\n",
    "        .option(\"checkpointLocation\", \"checkpoint\") \\\n",
    "        .start()\n",
    "    return write_query"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e4cb2140-9f2e-4914-b74c-be4c18cdbe8a",
   "metadata": {},
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "63abe115-879c-4863-97d3-b22cda7f7469",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
